//
// Created by martin on 8/19/21.
//

#include "nthread.h"
#include "error.h"

#include <stdlib.h>
#include <stdio.h>
#include <unistd.h>
#include <event2/event.h>

static int nthread_setup(nThread *thread);

/**
 * 收到管道消息时，调用该回调函数，从任务队列取出一个待处理任务，进行初始化
 * @param fd 监听的发生事件的文件描述符
 * @param what 事件类型，值为EV_READ 或 EV_WRITE
 * @param arg 用户参数
 */
static void notify_cb(evutil_socket_t fd, short what, void *arg)
{
    nThread *thread = (nThread *)arg;
    Task *task = NULL;

    printf("thread %d receive activation notice\n", thread->id);

    // 如果是读消息，则从待处理队列取出一个任务并初始化
    if (what & EV_READ) {
        char buf[2];
        // 从fd读取1byte数据
        if (read(fd, buf, 1) <= 0) { /* 从fd读取消息出错，或没有读取到数据 */
            return;
        }
        // 从fd成功读取数据
        // 从线程待处理任务队列头部取出一个Task，对齐进行初始化
        pthread_mutex_lock(&thread->mutex);
        task = queue_poll(thread->taskq);
        pthread_mutex_unlock(&thread->mutex);

        // 初始化任务
        if (task) {
            task_init(task, thread->base);
        }
    }
}

/**
 * 子线程主函数
 * @param arg 用户参数
 * @return void
 */
void *thread_main(void *arg)
{
    nThread *thread = (nThread *)arg;

    printf("thread_main %d begin\n", thread->id);

    // 事件循环
    event_base_dispatch(thread->base);
    // 释放libevent上下文资源
    event_base_free(thread->base);

    printf("thread_main %d end\n", thread->id);

    return NULL;
}

/**
 * 创建线程对象
 * @return 创建结果
 * - NULL 创建失败
 * - Not-NULL 指向新建线程对象的指针
 */
nThread *nthread_new()
{
    nThread *thread = (nThread *)malloc(sizeof(nThread));
    if (!thread) {
        err_sys("malloc error");
        return NULL;
    }

    return thread;
}

/**
 * 释放线程对象
 * @param thread 线程对象
 * @return void
 */
void nthread_free(nThread *thread)
{
    if (!thread)  return;

    if (thread->taskq)
        queue_free(thread->taskq);

    // 销毁互斥量
    pthread_mutex_destroy(&thread->mutex);
    // 释放nThread对象空间
    free(thread);
}

/**
 * 分离线程的创建和初始化，目地在于方便调用者创建线程数组，以及何时决定初始化、激活线程
 * @param thread 指向一个线程对象
 * @param id 调用者指定的编号
 * @return void
 */
void nthread_init(nThread *thread, int id)
{
    if (!thread) return;

    // 初始化任务队列
    thread->taskq = queue_new();
    if (!thread->taskq) {
        err_sys("create task queue failed");
    }

    // 初始化线程id
    thread->id = id;

    // 初始化线程属性
    pthread_attr_t attr;
    int err;
    err = pthread_attr_init(&attr);
    if (err != 0) {
        err_exit(err, "pthread_attr_init error");
    }

    /* 设置线程分离属性, 分离线程运行结束后自动释放资源，无需其他线程连接 */
    err = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
    if (err != 0) {
        err_exit(err, "pthread_attr_setdetachstate error");
        pthread_attr_destroy(&attr);
    }

    // 创建互斥量
    pthread_mutex_init(&thread->mutex, NULL);

    // 安装线程
    nthread_setup(thread);

    // 创建线程
    // 线程创建后就会立即运行
    pthread_create(&thread->tid, &attr, thread_main, thread);
    if (err != 0) {
        err_exit(err, "pthread_create error");
    }
}

/**
 * 向线程任务队列添加任务
 * @param thread 线程对象
 * @param task 新的待插入任务，task指向对象需要存储在堆中
 * @return
 * - -1 失败
 * - 0 成功
 */
int nthread_addTask(nThread *thread, Task *task)
{
    if (!thread || !task) return -1;

    pthread_mutex_lock(&thread->mutex);
    queue_offer(thread->taskq, task); /* 插入一个任务 */
    pthread_mutex_unlock(&thread->mutex);

    return 0;
}

/**
 * 安装线程，为线程创建libevent上下文
 * @param thread 线程对象
 * @return 安装结果
 * - -1 失败
 * - 0 成功
 * @details 为线程创建libevent上下文，添加监听添加用户任务Task的管道文件描述符。当主线程添加用户任务时，线程event_base接收到管道消息，
 * 进而在回调函数中初始化任务Task
 * @note 每个线程维护一个event_base，监听多个客户连接等事件
 */
int nthread_setup(nThread *thread)
{
    if (!thread) return -1;

    int fds[2];
    // fds[0] 绑定到写端；fds[1] 绑定到读端
    if (pipe(fds) < 0) {
        err_sys("pipe error");
    }

    // 读端绑定到event事件，写端fds[1]保存到notify_send_fd
    thread->notify_send_fd = fds[1];

    /* 创建libevent上下文
     * 添加管道监听事件，用于激活线程执行
     * 无锁是因为后续监听事件、执行事件处理任务（回调），都是在单线程中执行的
     */
    struct event_base *base;
    struct event_config *conf;

    // 创建libevent上下文
    conf = event_config_new();
    event_config_set_flag(conf, EVENT_BASE_FLAG_NOLOCK);
    base = event_base_new_with_config(conf);
    event_config_free(conf);

    if (!base) {
        err_msg("event_base_new_with_config error");
        return -1;
    }
    thread->base = base;

    // 监听管道事件
    struct event *ev = event_new(base, fds[0], EV_READ | EV_PERSIST, notify_cb, thread);
    event_add(ev, NULL);

    return 0;
}

/**
 * 激活线程，触发从任务队列取出一个任务并初始化
 * @param thread 线程对象
 * @return 激活结果
 * - -1 失败
 * - 0 成功
 * @note 通过管道信号激活线程目的何在？为什么不直接初始化任务并执行？
 * 因为只有正要处理的任务，才有必要初始化，因而任务的初始化交给了每个线程来处理，而线程当前可能正在执行其他任务。
 * 也就是说，监听到客户连接并创建任务、激活线程，与线程的执行是异步的。
 */
int nthread_activate(nThread *thread)
{
    if (!thread) return -1;

    printf("activate thread %d\n", thread->id);

    ssize_t ret = write(thread->notify_send_fd, "c", 1);
    if (ret < 0) {
        err_ret("activate error");
        return -1;
    }

    return 0;
}